package cn.doitedu.kafka.day01

import java.util.Properties

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import org.apache.kafka.common.serialization.StringSerializer

/**
 * 使用scala编写Kafka的生产者
 *
 * 1.Producer是将数据发送到哪个分区中呢？
 * 默认是轮询写入
 * 2.可不可以指定将数据发送到指定的分区中
 * 可以 new ProducerRecord[String, String](topic, 2, null, "my data : " + 888888)
 * 3.将固定key的数据，写入到同一个分区中
 * 底层使用的类似hashPartitioner，将key的hashcode值去非负数的模
 */
object ProducerDemo4 {

  def main(args: Array[String]): Unit = {


    //指定生产者的参数
    val properties = new Properties()
    //broker的地址
    // 连接kafka节点
    properties.setProperty("bootstrap.servers", "node-1.51doit.cn:9092,node-2.51doit.cn:9092,node-3.51doit.cn:9092")
    //指定key序列化方式
    properties.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    //指定value序列化方式
    properties.setProperty("value.serializer", classOf[StringSerializer].getName) // 两种写法都行

    //topic名称
    val topic = "test"
    //创建Kafka生产者
    val producer = new KafkaProducer[String, String](properties)

    //创建ProducerRecord时，指定分区编号，即将数据发送到指定的分区

    val record1 = new ProducerRecord[String, String](topic, "spark", "my data : " + 88888888)
    //发送数据
    producer.send(record1)

    val record2 = new ProducerRecord[String, String](topic, "spark", "my data : " + 7777777)
    //发送数据
    producer.send(record2)

    val record3 = new ProducerRecord[String, String](topic, "hive", "my data : " + 666666)
    //发送数据
    producer.send(record3)

    val record4 = new ProducerRecord[String, String](topic, "hive", "my data : " + 555555)
    //发送数据
    producer.send(record4)


    producer.close()


  }

}
